// 19_tokio.rs
// Rust Tokio 异步运行时详细教程

//! Tokio 是 Rust 生态中最流行的异步运行时，提供了构建异步应用程序所需的基础设施。
//! 本教程将详细介绍 Tokio 的各种功能和用法，包括：
//! 1. 基本异步任务执行
//! 2. 任务调度和并发控制
//! 3. 异步 I/O 操作
//! 4. 同步原语
//! 5. 定时器和时间处理
//! 6. 流处理
//! 7. 错误处理
//! 8. 性能监控和调试

use tokio::time::{sleep, Duration};
use tokio::sync::{mpsc, Mutex, Semaphore};
use std::sync::Arc;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Tokio 异步运行时详细教程 ===");
    
    // 1. 基本异步任务执行
    println!("\n1. 基本异步任务执行");
    basic_async_tasks().await;
    
    // 2. 任务调度和并发控制
    println!("\n2. 任务调度和并发控制");
    task_scheduling().await;
    
    // 3. 异步 I/O 操作
    println!("\n3. 异步 I/O 操作");
    async_io_operations().await;
    
    // 4. 同步原语
    println!("\n4. 同步原语");
    synchronization_primitives().await;
    
    // 5. 定时器和时间处理
    println!("\n5. 定时器和时间处理");
    timers_and_time_handling().await;
    
    // 6. 流处理
    println!("\n6. 流处理");
    stream_processing().await;
    
    // 7. 错误处理
    println!("\n7. 错误处理");
    error_handling().await;
    
    // 8. 性能监控和调试
    println!("\n8. 性能监控和调试");
    performance_monitoring().await;
    
    println!("\nTokio 教程完成！");
    Ok(())
}

// 1. 基本异步任务执行
async fn basic_async_tasks() {
    println!("--- 基本异步任务执行 ---");
    
    // 简单的异步函数调用
    let result = simple_async_function().await;
    println!("简单异步函数结果: {}", result);
    
    // 并发执行多个任务
    let task1 = tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        "任务1完成"
    });
    
    let task2 = tokio::spawn(async {
        sleep(Duration::from_millis(200)).await;
        "任务2完成"
    });
    
    let (result1, result2) = tokio::join!(task1, task2);
    println!("并发任务结果: {}, {}", result1.unwrap(), result2.unwrap());
    
    // 顺序执行任务
    let result1 = simple_async_function().await;
    let result2 = another_async_function().await;
    println!("顺序执行结果: {}, {}", result1, result2);
}

async fn simple_async_function() -> i32 {
    sleep(Duration::from_millis(50)).await;
    42
}

async fn another_async_function() -> String {
    sleep(Duration::from_millis(30)).await;
    "Hello, Tokio!".to_string()
}

// 2. 任务调度和并发控制
async fn task_scheduling() {
    println!("--- 任务调度和并发控制 ---");
    
    // 使用 spawn 创建任务
    let handles: Vec<_> = (0..5)
        .map(|i| {
            tokio::spawn(async move {
                sleep(Duration::from_millis(100)).await;
                format!("任务 {} 完成", i)
            })
        })
        .collect();
    
    // 等待所有任务完成
    for handle in handles {
        println!("{}", handle.await.unwrap());
    }
    
    // 限制并发数量
    let semaphore = Arc::new(Semaphore::new(3)); // 最多3个并发
    let mut handles = Vec::new();
    
    for i in 0..10 {
        let semaphore = semaphore.clone();
        let handle = tokio::spawn(async move {
            let _permit = semaphore.acquire().await.unwrap();
            sleep(Duration::from_millis(50)).await;
            format!("受限制的任务 {} 完成", i)
        });
        handles.push(handle);
    }
    
    for handle in handles {
        println!("{}", handle.await.unwrap());
    }
}

// 3. 异步 I/O 操作
async fn async_io_operations() {
    println!("--- 异步 I/O 操作 ---");
    
    // 模拟异步文件操作
    let file_content = async_file_read("example.txt").await;
    println!("读取文件内容: {}", file_content);
    
    // 模拟异步网络请求
    let response = async_http_request("https://httpbin.org/delay/1").await;
    println!("HTTP 请求响应: {}", response);
}

async fn async_file_read(filename: &str) -> String {
    // 模拟文件读取延迟
    sleep(Duration::from_millis(100)).await;
    format!("读取了文件 {} 的内容", filename)
}

async fn async_http_request(url: &str) -> String {
    // 模拟网络请求延迟
    sleep(Duration::from_millis(200)).await;
    format!("GET {} 响应: 200 OK", url)
}

// 4. 同步原语
async fn synchronization_primitives() {
    println!("--- 同步原语 ---");
    
    // 使用 Mutex 共享可变状态
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];
    
    for _ in 0..10 {
        let counter = counter.clone();
        let handle = tokio::spawn(async move {
            let mut num = counter.lock().await;
            *num += 1;
            sleep(Duration::from_millis(10)).await;
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.await.unwrap();
    }
    
    println!("计数器最终值: {}", *counter.lock().await);
    
    // 使用通道进行任务间通信
    let (tx, mut rx) = mpsc::channel(32);
    
    // 发送者任务
    let tx1 = tx.clone();
    let sender = tokio::spawn(async move {
        for i in 0..5 {
            if let Err(_) = tx1.send(format!("消息 {}", i)).await {
                break;
            }
            sleep(Duration::from_millis(50)).await;
        }
    });
    
    // 接收者任务
    let _receiver = tokio::spawn(async move {
        while let Some(msg) = rx.recv().await {
            println!("接收到: {}", msg);
        }
    });
    
    // 等待发送者完成
    let _ = sender.await;
    // 给接收者一点时间处理消息
    sleep(Duration::from_millis(100)).await;
    // 注意：在实际应用中，我们需要更优雅地关闭接收者
}

// 5. 定时器和时间处理
async fn timers_and_time_handling() {
    println!("--- 定时器和时间处理 ---");
    
    // 基本延迟
    println!("开始延迟...");
    sleep(Duration::from_millis(100)).await;
    println!("延迟结束");
    
    // 定时重复任务
    let mut interval = tokio::time::interval(Duration::from_millis(200));
    for i in 0..5 {
        interval.tick().await;
        println!("定时任务执行第 {} 次", i + 1);
    }
    
    // 超时控制
    let result = tokio::time::timeout(
        Duration::from_millis(300),
        long_running_task()
    ).await;
    
    match result {
        Ok(_) => println!("长时任务在超时前完成"),
        Err(_) => println!("长时任务超时"),
    }
}

async fn long_running_task() {
    sleep(Duration::from_secs(1)).await;
}

// 6. 流处理
async fn stream_processing() {
    println!("--- 流处理 ---");
    
    // 创建一个异步流
    let mut counter = 0;
    while counter < 5 {
        sleep(Duration::from_millis(100)).await;
        counter += 1;
        println!("流数据: {}", counter);
    }
    
    // 使用通道模拟流处理
    let (tx, mut rx) = mpsc::channel(100);
    
    // 数据生产者
    let _producer = tokio::spawn(async move {
        for i in 0..10 {
            if let Err(_) = tx.send(i * i).await {
                // 通道已关闭，退出
                break;
            }
            sleep(Duration::from_millis(50)).await;
        }
    });
    
    // 数据消费者
    let mut sum = 0;
    while let Some(value) = rx.recv().await {
        sum += value;
        println!("接收到值: {}, 当前总和: {}", value, sum);
        if sum > 50 {
            break;
        }
    }
}

// 7. 错误处理
async fn error_handling() {
    println!("--- 错误处理 ---");
    
    // 处理可能失败的异步操作
    match fallible_async_operation(true).await {
        Ok(result) => println!("操作成功: {}", result),
        Err(e) => println!("操作失败: {}", e),
    }
    
    match fallible_async_operation(false).await {
        Ok(result) => println!("操作成功: {}", result),
        Err(e) => println!("操作失败: {}", e),
    }
    
    // 使用 ? 操作符传播错误
    if let Err(e) = function_with_error_propagation().await {
        println!("错误传播示例: {}", e);
    }
}

async fn fallible_async_operation(success: bool) -> Result<String, &'static str> {
    sleep(Duration::from_millis(100)).await;
    if success {
        Ok("操作成功完成".to_string())
    } else {
        Err("操作失败")
    }
}

async fn function_with_error_propagation() -> Result<(), Box<dyn std::error::Error>> {
    let result = fallible_async_operation(false).await?;
    println!("结果: {}", result);
    Ok(())
}

// 8. 性能监控和调试
async fn performance_monitoring() {
    println!("--- 性能监控和调试 ---");
    
    // 测量任务执行时间
    let start = tokio::time::Instant::now();
    
    let task = tokio::spawn(async {
        sleep(Duration::from_millis(200)).await;
        "耗时任务完成"
    });
    
    let result = task.await.unwrap();
    let duration = start.elapsed();
    
    println!("{} 耗时: {:?}", result, duration);
    
    // 使用任务本地存储
    tokio::task::LocalSet::new().run_until(async {
        tokio::task_local! {
            static REQUEST_ID: u32;
        }
        
        let handle = tokio::spawn(REQUEST_ID.scope(42, async {
            println!("任务中的请求ID: {}", REQUEST_ID.get());
        }));
        
        handle.await.unwrap();
    }).await;
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[tokio::test]
    async fn test_simple_async_function() {
        let result = simple_async_function().await;
        assert_eq!(result, 42);
    }
    
    #[tokio::test]
    async fn test_fallible_operation_success() {
        let result = fallible_async_operation(true).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "操作成功完成");
    }
    
    #[tokio::test]
    async fn test_fallible_operation_failure() {
        let result = fallible_async_operation(false).await;
        assert!(result.is_err());
        assert_eq!(result.err().unwrap(), "操作失败");
    }
    
    #[tokio::test]
    async fn test_counter_with_mutex() {
        let counter = Arc::new(Mutex::new(0));
        let mut handles = vec![];
        
        for _ in 0..10 {
            let counter = counter.clone();
            let handle = tokio::spawn(async move {
                let mut num = counter.lock().await;
                *num += 1;
            });
            handles.push(handle);
        }
        
        for handle in handles {
            handle.await.unwrap();
        }
        
        assert_eq!(*counter.lock().await, 10);
    }
}